package com.ahui.test;

import com.athui.bean.ColumnMetaData;
import com.athui.bean.DataType;
import com.athui.enums.DataRegularEnums;
import com.athui.service.QueueExplorerService;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * @description: TODO
 * @autor: zhangzhonghui
 * @create: 2024-08-20 17:02
 * @Version: 1.0
 */
public class StringQueueExplorerService2 extends QueueExplorerService<String,Tuple2<String,String>> {


    public StringQueueExplorerService2(List<String> dataList) {
        super(dataList);
    }


    @Override
    public Tuple2<String, String> transformBySourceType(String input) {
        String[] split = input.split(",");
        if (split.length ==1) {
            return Tuple2.of(split[0], "");
        }
        return Tuple2.of(split[0], split[1]);
    }



    @Override
    public Object transformByValue(String index, Tuple2<String, String> row) {
        if ("0".equals(index)){
            return row.f0;
        }else if ("1".equals(index)){
            return row.f1;
        }
        return null;
    }

    @Override
    public Collection<ColumnMetaData> getColumnMetaDataList(Collection<ColumnMetaData> collect) {
        ColumnMetaData phone = new ColumnMetaData();
        phone.setColumnCode("phone");
        phone.setColumnName("phone");
        phone.setTypeLabel(DataType.TEXT);
        phone.setIndex("0");
        phone.setInvolvedRange("[13000008916x]\\&\\&[13000008916]");
        phone.setTextPattern(DataRegularEnums.TEXT_PHONE_REGULAR.getValue());
        phone.setFlinkColumnType(BasicTypeInfo.STRING_TYPE_INFO);
        collect.add(phone);

        ColumnMetaData address = new ColumnMetaData();
        address.setColumnCode("address");
        address.setColumnName("address");
        address.setTypeLabel(DataType.TEXT);
        address.setIndex("1");
        address.setFlinkColumnType(BasicTypeInfo.STRING_TYPE_INFO);
        collect.add(address);

        return collect;
    }

    @Override
    public String getJobName() {
        return "string-queue-job";
    }

    @Override
    public void sink(DataStream<ColumnMetaData> stream) {
        stream.print();
    }

    public static void main(String[] args) {

        List<String> dataList = new ArrayList<>();
        dataList.add("13000008918,北京东路1号");
        dataList.add("13000008917,北京东路2号");
        dataList.add("13000008916,");
        dataList.add("13000008915,北京东路4号");
        dataList.add("13000008914,北京东路5号");
        dataList.add("13000008913,北京东路6号");
        dataList.add("13000008912,北京东路7号");
        dataList.add("13000008911,北京东路8号");
        dataList.add("13000008910,北京东路9号");
        dataList.add("asgjkaljdcklas,北京东路10号");
        dataList.add("31241233414312,北京东路11号");

        StringQueueExplorerService2 explorerService =new StringQueueExplorerService2(dataList);

        try {
            explorerService.run();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


}
